/*
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package io.trino.jdbc.tuple.rs;

import io.trino.jdbc.tuple.protocol.FieldsProtocolType;
import lombok.Getter;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.math.NumberUtils;

import java.math.BigDecimal;
import java.sql.Time;
import java.sql.Timestamp;
import java.sql.Types;
import java.util.Date;
import java.util.Optional;
import java.util.function.Function;

/**
 * MYSQL 支持的数据类型 DataTypeEnum
 *
 * @see
 */
public enum DataType implements IDataType {
    // ##########################################
    // 数值类型
    // ##########################################
    /**
     * 值只能是0或1，存储字节数1比特位。
     * 取值0的逻辑意义为假。  取值1的逻辑意义为真。
     */
    BOOLEAN("boolean", Types.BOOLEAN, FieldsProtocolType.FIELD_TYPE_TINY,
            (IDataConver) source -> {
                if(source == null){
                    return Boolean.FALSE;
                }
                if (source instanceof Boolean) {
                    return ((Boolean)source).booleanValue();
                }

                String val = source.toString();
                if(StringUtils.equalsIgnoreCase(val, "true")){
                    return Boolean.TRUE;
                }
                if(StringUtils.equalsIgnoreCase(val, "false")){
                    return Boolean.FALSE;
                }

                // 0, 1 不转
                return Boolean.getBoolean(val);
            }),
    /**
     * BYTE, also is TINYINT
     * <p>
     * tinyint微整数类型	取值范围-128~127，存储字节数1字节。
     */
    BYTE("tinyint", Types.TINYINT, FieldsProtocolType.FIELD_TYPE_TINY, (IDataConver) source -> {
        if(source == null){
            return null;
        }

        String val = source.toString();
        if (source instanceof Integer) {
            return ((Integer)source).intValue();
        }

        // 不是数字
        if(!NumberUtils.isCreatable(val)){
            return null;
        }

        return NumberUtils.createInteger(val);
    }),
    /**
     * smallint小整数类型	取值范围-32768~32767，存储字节数2字节。
     */
    SHORT("smallint", Types.SMALLINT, FieldsProtocolType.FIELD_TYPE_SHORT, (IDataConver) source -> {
        if(source == null){
            return null;
        }

        String val = source.toString();
        if (source instanceof Integer) {
            return ((Integer)source).intValue();
        }

        // 不是数字
        if(!NumberUtils.isCreatable(val)){
            return null;
        }

        return NumberUtils.createInteger(val);
    }),
    /**
     * mediumint: 取值范围-8388608~8388607，存储字节数3字节。
     */
    MEDIUMINT("mediumint", Types.INTEGER, FieldsProtocolType.FIELD_TYPE_LONG, (IDataConver) source -> {
        if (source == null) {
            return null;
        }

        String val = source.toString();
        if (source instanceof Integer) {
            return ((Integer) source).intValue();
        }

        // 不是数字
        if (!NumberUtils.isCreatable(val)) {
            return null;
        }

        return NumberUtils.createInteger(val);
    }),
    /**
     * int或integer整数类型	取值范围-2147483648~2147483647，存储字节数4字节。
     */
    INT("int", Types.INTEGER, FieldsProtocolType.FIELD_TYPE_LONG, (IDataConver) source -> {
        if(source == null){
            return null;
        }

        String val = source.toString();
        if (source instanceof Integer) {
            return ((Integer)source).intValue();
        }

        // 不是数字
        if(!NumberUtils.isCreatable(val)){
            return null;
        }

        return NumberUtils.createInteger(val);
    }),
    /**
     * bigint大整数类型	取值范围-9223372036854775808~9223372036854775807，存储字节数8字节。
     */
    LONG("bigint", Types.BIGINT, FieldsProtocolType.FIELD_TYPE_LONGLONG, (IDataConver) source -> {
        if(source == null){
            return null;
        }

        String val = source.toString();
        if (source instanceof Long) {
            return ((Long)source).longValue();
        }

        // 不是数字
        if(!NumberUtils.isCreatable(val)){
            return null;
        }

        return NumberUtils.createLong(val);
    }),
    /**
     * float单精度浮点数	取值范围-3.402823466E+38~-1.175494351E-38, 0, 1.175494351E-38~3.402823466E+38，
     * IEEE标准，存储字节数4字节。
     */
    FLOAT("float", Types.FLOAT, FieldsProtocolType.FIELD_TYPE_FLOAT, (IDataConver) source -> {
        if(source == null){
            return null;
        }

        String val = source.toString();
        if (source instanceof Float) {
            return ((Float)source).floatValue();
        }

        // 不是数字
        if(!NumberUtils.isCreatable(val)){
            return null;
        }

        return NumberUtils.createFloat(val);
    }),
    /**
     * double双精度浮点数	取值范围-1.7976931348623157E+308~-2.2250738585072014E-308, 0, 2.2250738585072014E-308~1.7976931348623157E+308，
     * IEEE标准，存储字节数8字节。
     */
    DOUBLE("double", Types.DOUBLE, FieldsProtocolType.FIELD_TYPE_DOUBLE, (IDataConver) source -> {
        if(source == null){
            return null;
        }

        String val = source.toString();
        if (source instanceof Double) {
            return ((Double)source).doubleValue();
        }

        // 不是数字
        if(!NumberUtils.isCreatable(val)){
            return null;
        }

        return NumberUtils.createDouble(val);
    }),
    /**
     * decimal(m,d)或numeric.
     * m是数值的最大精度，取值范围为1~1000；  d是小数点右侧数字的位数，要求d≤m。
     * <p>
     * MySQL支持的最大精度为65. SQL网关支持最大 ???
     */
    DECIMAL("decimal", Types.DECIMAL, FieldsProtocolType.FIELD_TYPE_NEW_DECIMAL, (IDataConver) source -> {
        if(source == null){
            return null;
        }

        String val = source.toString();
        if (source instanceof BigDecimal) {
            return (BigDecimal) source;
        }

        // 不是数字
        if(!NumberUtils.isCreatable(val)){
            return null;
        }

        return NumberUtils.createBigDecimal(val);
    }),
    NUMERIC("numeric", Types.NUMERIC, FieldsProtocolType.FIELD_TYPE_NEW_DECIMAL, (IDataConver) source -> {
        return DECIMAL.conver.get().conver(source);
    }),

    /**
     * time时间类型
     * 取值范围'00:00:00'~'23:59:59'，支持的数据格式为'HH:MM:SS'，存储字节数为8字节。
     *
     * <pre>
     *      MySQL支持自定义精度。
     *      ADB支持的精确为毫秒，即小数点后三位。
     * </pre>
     */
    TIME("time", Types.TIME, FieldsProtocolType.FIELD_TYPE_TIME, (IDataConver) source -> {
        if(source == null){
            return null;
        }

        String val = source.toString();
        if (source instanceof Time) {
            return (Time) source;
        }

        java.util.Date date = null;
        if (source instanceof Date) {
            date = (Date) source;
        }else{
            try {
                date = TIME_SDF.parse(val);
            } catch (Exception e) {
                try {
                    date = DATE_TIME_SDF.parse(val);
                } catch (Exception ex) {
                    // ignore
                }
            }
        }

        if(date == null){
            return null;
        }

        Time time = new java.sql.Time(date.getTime());
        // Time time1 = time.valueOf(val);
        return time;
    }),
    /**
     * date日期类型
     * 取值范围'0001-01-01'~'9999-12-31'，支持的数据格式为'YYYY-MM-DD'，存储字节数为4字节。
     *
     * <pre>
     *      MySQL支持0000-00-00。
     *      ADB对时间类型的数值会进行合法性校验。如果开启参数ILLEGAL_DATE_CONVERT_TO_NULL_ENABLE=true，
     *      在不合理的数值写入时， 例如0000-00-00，ADB会自动将其转化为NULL。请确保写入的日期和时间有意义。
     * </pre>
     */
    DATE("date", Types.DATE, FieldsProtocolType.FIELD_TYPE_DATE, (IDataConver) source -> {
        if(source == null){
            return null;
        }

        String val = source.toString();
        if (source instanceof java.sql.Date) {
            return (java.sql.Date) source;
        }

        java.util.Date date = null;
        if (source instanceof Date && !(source instanceof Time)) {
            date = (Date) source;
        }else{
            try {
                date = DATE_SDF.parse(val);
            } catch (Exception e) {
                try {
                    date = DATE_TIME_SDF.parse(val);
                } catch (Exception ex) {
                    // ignore
                }
            }
        }

        if(date == null){
            return null;
        }

        java.sql.Date d = new java.sql.Date(date.getTime());
        return d;
    }),
    /**
     * 时间戳类型，取值范围'0001-01-01 00:00:00.000'UTC~'9999-12-31 23:59:59.999'UTC，
     * 支持的数据格式为'YYYY-MM-DD HH:MM:SS'，存储字节数为4字节。
     *
     * <pre>
     *      MySQL支持自定义精度。
     *      支持的精确为毫秒，即小数点后三位。
     * </pre>
     */
    TIMESTAMP("timestamp", Types.TIMESTAMP, FieldsProtocolType.FIELD_TYPE_TIMESTAMP, (IDataConver) source -> {
        if(source == null){
            return null;
        }

        String val = source.toString();
        if (source instanceof Timestamp) {
            return (Timestamp) source;
        }

        java.util.Date date = null;
        if (source instanceof Date && !(source instanceof Time)) {
            date = (Date) source;
        }else{
            try {
                date = TIMESTAMP_SDF.parse(val);
            } catch (Exception e) {
                try {
                    date = DATE_TIME_SDF.parse(val);
                } catch (Exception ex) {
                    try {
                        date = DATE_SDF.parse(val);
                    } catch (Exception ee) {
                        // ignore
                    }
                }
            }
        }
        if(date == null){
            return null;
        }

        Timestamp timestamp = new Timestamp(date.getTime());
        return timestamp;
    }),
    /**
     * 取值范围'0001-01-01 00:00:00.000'UTC~'9999-12-31 23:59:59.999'UTC，
     * 支持的数据格式为'YYYY-MM-DD HH:MM:SS'，存储字节数为8字节。
     *
     * <pre>
     *      MySQL支持0000-00-00。
     * </pre>
     */
    DATETIME("datetime", Types.TIMESTAMP, FieldsProtocolType.FIELD_TYPE_DATETIME, (IDataConver) source -> {
        return TIMESTAMP.conver.get().conver(source);
    }),

    /**
     * varchar变长字符串类型	存储字节数最大为16MB，使用时无需指定存储长度。
     * <p>
     * MySQL中的char、varchar、text、mediumtext或者longtext。
     */
    STRING("varchar", Types.VARCHAR, FieldsProtocolType.FIELD_TYPE_VARCHAR, (IDataConver) source -> {
        if(source == null){
            return null;
        }

        if (source instanceof String) {
            return (String) source;
        }

        String val = source.toString();
        return val;
    }),


    // for blob and binary, for calcite SqlTypeName not support blob
    // for object
    NULL("null", Types.NULL, FieldsProtocolType.FIELD_TYPE_NULL, (IDataConver) source -> {
        if(source == null){
            return null;
        }

        String val = source.toString();
        if(StringUtils.equalsIgnoreCase(val, "null")){
            return null;
        }

        return source;
    }),
    //    BYTES(13, "bytes", Types.BINARY),
    @Deprecated
    OTHER("other", Types.OTHER, FieldsProtocolType.FIELD_TYPE_STRING),
    @Deprecated
    MULTIVALUE("multivalue", Types.VARCHAR),
    @Deprecated
    GEO2D("geo2d", Types.VARCHAR),
    @Deprecated
    BLOB("blob", Types.BLOB),
    /**
     * binary二进制字符串类型	存储字符长度
     */
    @Deprecated
    BINARY("binary", Types.BINARY),

    // for array
    @Deprecated
    BYTE_ARRAY("array<tinyint>", Types.ARRAY),
    @Deprecated
    SHORT_ARRAY("array<smallint>", Types.ARRAY),
    @Deprecated
    INT_ARRAY("array<integer>", Types.ARRAY),
    @Deprecated
    FLOAT_ARRAY("array<float>", Types.ARRAY),

    // for point
    @Deprecated
    POINT("point", Types.VARCHAR),

    // for map
    @Deprecated
    MAP("map", Types.VARCHAR),

    // for json
    @Deprecated
    JSON("json", Types.VARCHAR),

    ;

    @Getter
    private String sqlName;

    /**
     * jdbc type, is java.sql.Types
     */
    @Getter
    private int sqlType;

    /**
     * MYSQL FIELD PACKET 协议中的 protocolType 类型
     */
    @Getter
    private FieldsProtocolType protocolType;

    private Optional<IDataConver> conver;

    private Optional<Function<Object, Object>> function;

    DataType(String sqlName, int sqlType) {
        this.sqlName = sqlName;
        this.sqlType = sqlType;
        this.protocolType = null;
        this.conver = Optional.empty();
    }

    DataType(String sqlName, int sqlType, FieldsProtocolType protocolType) {
        this.sqlName = sqlName;
        this.sqlType = sqlType;
        this.protocolType = protocolType;
        this.conver = Optional.empty();
    }

    DataType(String sqlName, int sqlType, FieldsProtocolType protocolType, IDataConver conver) {
        this.sqlName = sqlName;
        this.sqlType = sqlType;
        this.protocolType = protocolType;
        this.conver = Optional.of(conver);
    }

    DataType(String sqlName, int sqlType, FieldsProtocolType protocolType, Function<Object, Object> function) {
        this.sqlName = sqlName;
        this.sqlType = sqlType;
        this.protocolType = protocolType;
        this.function = Optional.of(function);
    }

    /**
     * 得到类型的实际值
     */
    public Object getSourceActualyValue(Object source){
        if(conver != null && conver.isPresent()){
            return conver.get().conver(source);
        }

        if(function != null && function.isPresent()){
            return function.get().apply(source);
        }

        return source;
    }

    // ##########################################################
    //  转换
    // ##########################################################
    /**
     * sqlType 转 core.DataType
     *
     * @param sqlType
     * @return DataType
     */
    public static DataType getBySqlType(int sqlType) {
        switch (sqlType) {
            case Types.CHAR:
            case Types.NCHAR:
            case Types.VARCHAR:
            case Types.LONGVARCHAR:
            case Types.NVARCHAR:
            case Types.LONGNVARCHAR:
                return STRING;
            case Types.NUMERIC:
                return NUMERIC;
            case Types.DECIMAL:
                return DECIMAL;
            case Types.BIT:
            case Types.BOOLEAN:
                return BOOLEAN;
            case Types.INTEGER:
                return INT;
            case Types.SMALLINT:
                return SHORT;
            case Types.TINYINT:
                return BYTE;
            case Types.BIGINT:
                return LONG;
            case Types.REAL:
            case Types.FLOAT:
                return FLOAT;
            case Types.DOUBLE:
                return DOUBLE;
            case Types.DATE:
                return DATE;
            case Types.TIME:
            case Types.TIME_WITH_TIMEZONE:
                return TIME;
            case Types.TIMESTAMP:
            case Types.TIMESTAMP_WITH_TIMEZONE:
                return TIMESTAMP;
            case Types.NULL:
                return NULL;
//            case Types.BLOB:
//            case Types.BINARY:
//            case Types.VARBINARY:
//            case Types.LONGVARBINARY:
//                return BYTES;
            default:
                return null;
//                throw ErrorMSGException.of(ErrorCodez.INVALID_VALUE_TYPE, "type => " + sqlType);
        }
    }

    /**
     * sqlType name 转 core.DataType
     *
     * @param name
     * @return DataType
     */
    public static DataType getBySqlType(String name) {
        if (name == null || name.length() <= 0) {
            return null;
        } else if (name.equalsIgnoreCase("bool")
                || name.equalsIgnoreCase("boolean")) {
            return DataType.BOOLEAN;
        } else if (name.equalsIgnoreCase("tinyint")
                || name.equalsIgnoreCase("byte")) {
            return DataType.BYTE;
        } else if (name.equalsIgnoreCase("smallint")
                || name.equalsIgnoreCase("short")) {
            return DataType.SHORT;
        } else if (name.equalsIgnoreCase(DataType.MEDIUMINT.sqlName)) {
            return DataType.MEDIUMINT;
        } else if (name.equalsIgnoreCase("int")
                || name.equalsIgnoreCase("integer")) {
            return DataType.INT;
        } else if (name.equalsIgnoreCase("bigint")
                || name.equalsIgnoreCase("long")) {
            return DataType.LONG;
        } else if (name.equalsIgnoreCase("double")) {
            return DataType.DOUBLE;
        } else if (name.equalsIgnoreCase("numeric")){
            return NUMERIC;
        } else if (name.equalsIgnoreCase("decimal")) {
            return DataType.DECIMAL;
        } else if (name.equalsIgnoreCase("real")
                || name.equalsIgnoreCase("float")) {
            return DataType.FLOAT;
        } else if (name.equalsIgnoreCase("time")) {
            return DataType.TIME;
        } else if (name.equalsIgnoreCase("date")) {
            return DataType.DATE;
        } else if (name.equalsIgnoreCase("timestamp")) {
            return DataType.TIMESTAMP;
        } else if (name.equalsIgnoreCase("datetime")) {
            return DataType.DATETIME;
        } else if (name.equalsIgnoreCase("varchar")
                || name.equalsIgnoreCase("string")) {
            return DataType.STRING;
        } else if (name.equalsIgnoreCase("multivalue")) {
            return DataType.MULTIVALUE;
        } else {
//            DataType arrayType = tryParseArrayType(name);
//            if (arrayType != null) {
//                return arrayType;
//            }

            return null;
//            throw ErrorMSGException.of(ErrorCodez.INVALID_VALUE_TYPE, "Invalid type name=>" + name);
        }
    }

    public static boolean isPremitiveArrayType(DataType type) {
        return type == DataType.BYTE_ARRAY || type == DataType.SHORT_ARRAY
                || type == DataType.INT_ARRAY || type == DataType.FLOAT_ARRAY;
    }

    /**
     * 是否为时间类型
     */
    public static boolean isDatetimeType(DataType type) {
        return type == DataType.DATE || type == DataType.TIME
                || type == DataType.TIMESTAMP || type == DataType.DATETIME;
    }

    /**
     * 是否为字符串
     */
    public static boolean isString(DataType type) {
        return type == DataType.STRING
                || type == DataType.GEO2D || type == DataType.MULTIVALUE
                || isPremitiveArrayType(type);
    }

    /**
     * 目前支持特定几种列数据类型, 不支持的类型返回null
     *
     * @param type
     * @return
     */
    @Deprecated
    public static DataType get(int type) {
        switch (type) {
            case 1:
                return BOOLEAN;
            case 2:
                return BYTE;
            case 3:
                return SHORT;
            case 4:
                return INT;
            case 5:
                return LONG;
            case 8: // 6
                return DECIMAL;
            case 7:
                return DOUBLE;
            case 6: // 8
                return FLOAT;
            case 9:
                return TIME;
            case 10:
                return DATE;
            case 11:
                return TIMESTAMP;
            case 14: // 13
                return STRING;
            case 12: // 14
                return DATETIME;
//            case 22:
//                return MULTIVALUE;
//            case 23:
//                return GEO2D;
//            case 12:
//                return BYTES;
//            case 24:
//                return BLOB;
//            case 25:
//                return BINARY;
            case 0:
                return NULL;
            case 19:
                return OTHER;
//            case 26:
//                return BYTE_ARRAY;
//            case 27:
//                return SHORT_ARRAY;
//            case 28:
//                return INT_ARRAY;
//            case 29:
//                return FLOAT_ARRAY;
        }

        return null;
//        throw ErrorMSGException.of(ErrorCodez.INVALID_VALUE_TYPE, "type => " + type);
    }

}
